package com.epoint.flinkdemo.utils;

import com.alibaba.druid.pool.DruidDataSource;
import org.apache.flink.api.java.utils.ParameterTool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

/**
 *
 * 数据源工具类
 * @author liufl
 * @version [版本号, 21-4-28]
 */
public class DataSourceUtils
{
    private volatile static DruidDataSource dataSource = null;

    private volatile static DruidDataSource ckDataSource = null;

    private volatile static ParameterTool parameterTool = null;

    private static Object lock1 = new Object();

    private static Object lock2 = new Object();

    private static Object lock3 = new Object();

    private static final Logger log = LoggerFactory.getLogger(DataSourceUtils.class);


    public static DruidDataSource getDataSource() {
        if (dataSource == null) {
            synchronized (lock1) {
                // 初始化MySQL, DruidDataSource
                dataSource = new DruidDataSource();
                dataSource.setUrl(getParameterTool().get("source.jdbcurl"));
                dataSource.setUsername(getParameterTool().get("source.username"));
                dataSource.setPassword(getParameterTool().get("source.password"));
                dataSource.setLogAbandoned(true);
                dataSource.setRemoveAbandoned(true);
                dataSource.setRemoveAbandonedTimeout(180);
            }
        }
        return dataSource;
    }

    public static DruidDataSource getCkDataSource() {
        if (ckDataSource == null) {
            synchronized (lock2) {
                ckDataSource = new DruidDataSource();
                ckDataSource.setUrl(getParameterTool().get("sink.jdbcurl"));
                ckDataSource.setUsername(getParameterTool().get("sink.username"));
                ckDataSource.setPassword(getParameterTool().get("sink.password"));
                ckDataSource.setLogAbandoned(true);
                ckDataSource.setRemoveAbandoned(true);
                ckDataSource.setRemoveAbandonedTimeout(180);
            }
        }
        return ckDataSource;
    }

    public static ParameterTool getParameterTool() {
        if (parameterTool == null) {
            synchronized (lock3) {
                try {
                    parameterTool = ParameterTool.fromPropertiesFile(DataSourceUtils.class.getResourceAsStream("/jdbc.properties"));
                } catch (IOException e) {
                    log.error("获取ParameterTool全局参数异常");
                }
            }
        }
        return parameterTool;
    }

    public static void closeDataSource(Connection connection, PreparedStatement ps, ResultSet resultSet) throws
            SQLException {
        if (resultSet != null) {
            resultSet.close();
        }
        if (ps != null) {
            ps.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}
